use std::time::Duration;

use libp2p::{futures::StreamExt, gossipsub::IdentTopic, kad::Mode, swarm::SwarmEvent};
use tokio::{select, sync::mpsc};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info};

use crate::{
    app::{
        GOSSIPSUB_PUBLIC_FILES_TOPIC_NAME, LOG_TARGET, P2pNetworkBehaviorEvent, P2pService,
        Service,
        store::{self},
    },
    error::{P2pNetworkError, ServerError},
};

#[async_trait::async_trait]
impl<FileStore> Service for P2pService<FileStore>
where
    FileStore: store::Store + Sync + Send + 'static + Clone,
{
    async fn start(&mut self, cancel_token: CancellationToken) -> Result<(), ServerError> {
        let mut swarm = self.swarm().await?;
        // 监听指定端口
        swarm
            .listen_on("/ip4/0.0.0.0/tcp/0".parse().map_err(|e| {
                ServerError::P2pNetwork(P2pNetworkError::Libp2pMultiaddrParseError(e))
            })?)
            .map_err(|e| ServerError::P2pNetwork(P2pNetworkError::Libp2pTransportError(e)))?;

        swarm
            .listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse().map_err(|e| {
                ServerError::P2pNetwork(P2pNetworkError::Libp2pMultiaddrParseError(e))
            })?)
            .map_err(|e| ServerError::P2pNetwork(P2pNetworkError::Libp2pTransportError(e)))?;

        swarm.behaviour_mut().kademlia.set_mode(Some(Mode::Server));

        // 节点在广域网中订阅该消息，以便接收其他节点发布的文件(需要做下处理，如果是自己发布的文件，自己也订阅了，那么需要过滤掉自己)
        let file_owners_topic = IdentTopic::new(GOSSIPSUB_PUBLIC_FILES_TOPIC_NAME);
        swarm
            .behaviour_mut()
            .gossipsub
            .subscribe(&file_owners_topic)
            .map_err(|e| {
                ServerError::P2pNetwork(P2pNetworkError::Libp2pGossipsubSubscriptionError(e))
            })?;

        // TODO: add bootstrap peers

        // start providing already download files from DB
        self.start_providing_all_files(&mut swarm);

        // start broadcasting periodically all published files
        let (publish_file_tx, mut publish_file_rx) = mpsc::channel(100);
        let file_store = self.file_store.clone();
        let cancel_token_clone = cancel_token.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(10));
            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        info!(target: LOG_TARGET, "Publishing all published files on DHT");
                        if let Ok(published_files) = file_store.fetch_all_public_published_files() {
                            for published_file in published_files {
                                let file_id = published_file.file_id;
                                match publish_file_tx.send(published_file).await {
                                    Ok(_) => info!(target: LOG_TARGET, "Published file: {file_id} on DHT"),
                                    Err(e) => error!(target: LOG_TARGET, "Error publishing file: {:?}", e),
                                }
                            }
                        } else {
                            error!(target: LOG_TARGET, "Error fetching published files from DB");
                        }
                    }
                    _ = cancel_token_clone.cancelled() => {
                        info!(target: LOG_TARGET, "P2P file broadcast service shutting down...");
                        break;
                    }
                }
            }
        });

        loop {
            select! {
                file_process_result = self.file_publish_receiver.recv() => {
                    if let Some(file_process_result) = file_process_result {
                        info!(target: LOG_TARGET, "New file: {:?} on DHT: {:?}", file_process_result.original_file_name, file_process_result.hash_id());
                        self.handle_file_publish(&mut swarm, file_process_result).await;
                    }
                },
                command = self.command_receiver.recv() => {
                    if let Some(command) = command {
                        info!(target: LOG_TARGET, "New command ->: {:?}", command);
                        self.handle_command(&mut swarm, command).await;
                    }
                }
                event = swarm.select_next_some() => match event {
                    SwarmEvent::Behaviour(event) => match event {
                        P2pNetworkBehaviorEvent::Identify(event) => self.process_identify_event(&mut swarm, event),

                        P2pNetworkBehaviorEvent::Ping(event) => {
                            info!(target: LOG_TARGET, "ping event {:?}",  event);
                        },
                        P2pNetworkBehaviorEvent::Mdns(event) => self.process_mdns_event(&mut swarm, event),
                        P2pNetworkBehaviorEvent::Kademlia(event) => {
                            self.process_kademlia_event(&mut swarm, event);
                        },
                        P2pNetworkBehaviorEvent::RelayServer(event) => {
                            debug!(target: LOG_TARGET, "relay server event {:?}",  event);
                        },
                        P2pNetworkBehaviorEvent::RelayClient(event) => {
                            debug!(target: LOG_TARGET, "relay client event {:?}",  event);
                        },
                        P2pNetworkBehaviorEvent::Gossipsub(event) => self.process_gossipsub_event(&mut swarm, event),
                        P2pNetworkBehaviorEvent::Dcutr(event) => {
                            debug!(target: LOG_TARGET, "dcutr event {:?}",  event);
                        },
                        P2pNetworkBehaviorEvent::FileDownload(event) => self.process_file_download_event(&mut swarm, event).await,
                        P2pNetworkBehaviorEvent::MetadataDownload(event) => self.process_metadata_download_event(&mut swarm, event).await,
                    },
                    SwarmEvent::NewListenAddr { listener_id: _listener_id, address } => {
                        info!(target: LOG_TARGET, "Listening on {:?}", address);
                    },

                    // SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, num_established, concurrent_dial_errors, established_in } => todo!(),
                    // SwarmEvent::ConnectionClosed { peer_id, connection_id, endpoint, num_established, cause } => todo!(),
                    // SwarmEvent::IncomingConnection { connection_id, local_addr, send_back_addr } => todo!(),
                    // SwarmEvent::IncomingConnectionError { connection_id, local_addr, send_back_addr, error } => todo!(),
                    // SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => todo!(),
                    // SwarmEvent::ExpiredListenAddr { listener_id, address } => todo!(),
                    // SwarmEvent::ListenerClosed { listener_id, addresses, reason } => todo!(),
                    // SwarmEvent::ListenerError { listener_id, error } => todo!(),
                    // SwarmEvent::Dialing { peer_id, connection_id } => todo!(),
                    // SwarmEvent::NewExternalAddrCandidate { address } => todo!(),
                    // SwarmEvent::ExternalAddrConfirmed { address } => todo!(),
                    // SwarmEvent::ExternalAddrExpired { address } => todo!(),
                    // SwarmEvent::NewExternalAddrOfPeer { peer_id, address } => todo!(),
                    _ => {
                        debug!(target: LOG_TARGET, "swarm event {:?}",  event);
                    },
                },

                Some(publish_file) = publish_file_rx.recv() => {
                    info!(target: LOG_TARGET, "Publishing file: {:?} on DHT", &publish_file.file_id);
                    self.handle_public_file(&mut swarm, &publish_file);
                }

                _ = cancel_token.cancelled() => {
                    info!(target: LOG_TARGET, "P2P networking service shutting down...");
                    break;
                }
            }
        }
        Ok(())
    }
}
